#include "RpcServer.h"
#include "MyRpc.h"
#include "MessageHeader.pb.h"
#include "ZkClient.h"
#include <iostream>
#include <thread>

void RpcServer::NotifyService(google::protobuf::Service *service)
{
    
    // std::cout << "RpcServer::NotifyService()" << std::endl;
    LOG_INFO("RpcServer::NotifyService()");

    ServiceInfo service_info;
    service_info.service = service;

    // 获取service描述信息
    const google::protobuf::ServiceDescriptor *service_desc = service->GetDescriptor();

    // service名字
    std::string service_name = service_desc->name();

    // 获取service的方法列表
    
    int method_cnt = service_desc->method_count();
    // std::cout << "service_name:" << service_name << " method_cnt:" << method_cnt << std::endl;
    LOG_INFO("service_name: %s method_cnt: %d", service_name.c_str(), method_cnt);
    for(int i = 0; i < method_cnt; i++)
    {
        // 获取服务中的每一个方法
        const google::protobuf::MethodDescriptor *method_desc = service_desc->method(i);
        std::string method_name = method_desc->name();
        // std::cout << "method_name:" << method_name << std::endl;
        LOG_INFO("method_name: %s", method_name.c_str());
        if(service_info.method_map.find(method_name) != service_info.method_map.end()) {
            // std::cout << "method_name:" << method_name << " is already exist" << std::endl;
            LOG_ERROR("method_name: %s is already exist", method_name.c_str());
            continue;
        } else {
            service_info.method_map[method_name] = method_desc;
        }
    }

    // 保存service及其方法列表
    m_service_map[service_name] = service_info;

    // 输出保存的内容
    for(auto &item : m_service_map)
    {
        // std::cout << "service_name:" << item.first << std::endl;
        LOG_INFO("service_name: %s", item.first.c_str());
        for(auto &method : item.second.method_map)
        {
            // std::cout << "method_name:" << method.first << std::endl;
            LOG_INFO("method_name: %s", method.first.c_str());
        }
    }
}

// 启动Rpc服务节点，开始提供Rpc远程网络调用服务
void RpcServer::Run()
{
    // std::cout << "RpcServer::Run()" << std::endl;
    LOG_INFO("RpcServer::Run()");

    // 设置ip地址和端口
    std::string ip = MyRpc::GetInstance().GetConfig().GetConfig("rpc_server_ip");
    uint16_t port = stoi(MyRpc::GetInstance().GetConfig().GetConfig("rpc_server_port"));
    muduo::net::InetAddress addr(ip, port);
    // std::cout << "server   ip:" << ip << " port:" << port << std::endl;
    LOG_INFO("server ip: %s port: %d", ip.c_str(), port);

    // 创建tcp服务对象
    muduo::net::TcpServer server(&m_event_loop_, addr, "RpcServer");

    // 绑定连接回调和消息读写回调
    server.setConnectionCallback(std::bind(&RpcServer::onConnection, this, std::placeholders::_1));
    server.setMessageCallback(std::bind(&RpcServer::onMessage, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));

    // 获取当前计算机上有多少个CPU核
    unsigned int hardware_threads = std::thread::hardware_concurrency();
    // std::cout << "Number of CPU cores: " << hardware_threads << std::endl;
    LOG_INFO("Number of CPU cores: %d", hardware_threads);

    // 设置muduo库线程数量
    server.setThreadNum(hardware_threads);

    // 将远程服务注册到zookeeper上
    std::string zk_ip = MyRpc::GetInstance().GetConfig().GetConfig("zookeeper_ip");
    uint16_t zk_port = stoi(MyRpc::GetInstance().GetConfig().GetConfig("zookeeper_port"));
    LOG_INFO("zookeeper ip: %s port: %d", zk_ip.c_str(), zk_port);
    ZkClient zk_client;
    zk_client.Connect(zk_ip, zk_port);
    // service_name 为永久性节点  method_name为临时性节点
    for(auto &service : m_service_map)
    {
        std::string service_name = service.first;
        zk_client.CreateNode("/" + service_name, "", false);
        for(auto &method : service.second.method_map)
        {
            std::string method_name = method.first;
            // 要记录的数据就是远程服务部署的ip地址和端口号
            std::string data = ip + ":" + std::to_string(port);
            zk_client.CreateNode("/" + service_name + "/" + method_name, data, true);
        }
    }

    // 启动网络服务
    LOG_INFO("RpcServer start");
    server.start();
    m_event_loop_.loop();
}

// 连接或断开回调
void RpcServer::onConnection(const muduo::net::TcpConnectionPtr &conn) 
{
    // 如果客户端断开连接，则关闭fd
    if (!conn->connected())
    {
        conn->shutdown();
    }
}

/*
buffer 格式
header_len（4Byte） + header_str  + args_str
header_len（4Byte）  作为数字可以方便的转换为二进制，没必要序列化

header_str 格式  （protobuf序列化为二进制）   protobuf文件由框架开发者提供，内容确定
service_name + method_name + args_size

args_str （protobuf序列化为二进制） protobuf文件由框架使用者提供，内容不确定
*/
// 消息读写回调
void RpcServer::onMessage(const muduo::net::TcpConnectionPtr &conn,
                          muduo::net::Buffer *buffer,
                          muduo::Timestamp time) 
{
    // 对buffer进行分割

    // 获取header_len
    std::string recv_buf = buffer->retrieveAllAsString();
    uint32_t header_len = 0;
    recv_buf.copy((char *)&header_len, 4, 0);

    // 获取header_str
    std::string header_str = recv_buf.substr(4, header_len);

    // 获取service_name method_name args_size
    std::string service_name="";
    std::string method_name="";
    uint32_t args_size=0;
    MessageHeader::MessageHeader message_header;
    if(message_header.ParseFromString(header_str) == false)
    {
        // std::cout << "message_header.ParseFromString() error" << std::endl;
        LOG_ERROR("message_header.ParseFromString() error");
        return;
    }
    else 
    {
        service_name = message_header.service_name();
        method_name = message_header.method_name();
        args_size = message_header.args_size();
        // std::cout << "service_name:" << service_name << " method_name:" << method_name << " args_size:" << args_size << std::endl;
        LOG_INFO("service_name: %s method_name: %s args_size: %d", service_name.c_str(), method_name.c_str(), args_size);
    }

    // 获取args_str （尚未反序列化）
    std::string args_str = recv_buf.substr(4 + header_len, args_size);

    // 获取service_info
    ServiceInfo *service_info = nullptr;
    google::protobuf::Service *service = nullptr;
    if(m_service_map.find(service_name) == m_service_map.end())
    {
        // std::cout << "service_name:" << service_name << " is not exist" << std::endl;
        LOG_ERROR("service_name: %s is not exist", service_name.c_str());
        return;
    }
    else 
    {
        service_info = &m_service_map[service_name];
        service = service_info->service;
    }

    // 获取method_desc
    const google::protobuf::MethodDescriptor *method_desc = nullptr;
    if(service_info->method_map.find(method_name) == service_info->method_map.end())
    {
        // std::cout << "method_name:" << method_name << " is not exist" << std::endl;
        LOG_ERROR("method_name: %s is not exist", method_name.c_str());
        return;
    }
    else 
    {
        method_desc = service_info->method_map[method_name];
    }

    // 对args_str进行反序列化，得到request
    google::protobuf::Message *request = service->GetRequestPrototype(method_desc).New();
    if(request->ParseFromString(args_str) == false)
    {
        // std::cout << "request->ParseFromString() error" << std::endl;
        LOG_ERROR("request->ParseFromString() error");
        return;
    }
    
    // 构建response
    google::protobuf::Message *response = service->GetResponsePrototype(method_desc).New();

    // 构建rpc方法回调函数
    google::protobuf::Closure *done = google::protobuf::NewCallback<RpcServer, const muduo::net::TcpConnectionPtr &, google::protobuf::Message*>(this, &RpcServer::sendRpcResponse, conn, response);

    // 调用rpc方法
    service->CallMethod(method_desc, nullptr, request, response, done);
}

// RPC方法调用回调
void RpcServer::sendRpcResponse(const muduo::net::TcpConnectionPtr &conn,
                                  google::protobuf::Message *response)
{
    std::string response_str = "";
    if(response->SerializeToString(&response_str))
    {
        conn->send(response_str);
    }
    else 
    {
        // std::cout << "response->SerializeToString() error" << std::endl;
        LOG_ERROR("response->SerializeToString() error");
    }

    // // 断开连接
    // conn->shutdown();
}